Cloud Composer で Dataplex のデータリネージ統合を試してみた

エノカワ

2024.09.16

こんにちは！エノカワです。

Cloud Composer は、Apache Airflow で構築されたフルマネージドのワークフローオーケストレーションサービスです。
Apache Airflow のオープンソースプロジェクトを基に構築されており、Python プログラミング言語を使用して動作します。

試したこと

今回は、Cloud Composer で Dataplex のデータリネージ統合を試してみました。
下記のドキュメントに沿って実施してみましたので、その内容をご紹介します。

Dataplex を使用したデータリネージ

Data Lineage API を有効にする

まずは Dataplex のデータリネージ機能を利用するために、Data Lineage API を有効にします。
APIの有効化は Google Cloud コンソールの 「APIとサービス」 メニューから行います。

cloud-composer-dataplex_04

Data Lineage API を有効にすると、Dataplex は BigQuery、Cloud Data Fusion、Dataproc のデータの取り込みを自動的に開始します。

環境作成

次に、Cloud Composer 環境を作成します。
Google Cloud コンソールの「Cloud Composer の環境作成」ページから、test-composerという名前の環境を、東京リージョンで作成しました。
サービスアカウントなどその他の設定はデフォルトのままで作成しました。

cloud-composer-dataplex_01

cloud-composer-dataplex_02

Cloud Composer 環境でデータリネージ統合を有効にする

Cloud Composer 環境でデータリネージ統合を有効にするには、Composer の環境設定を開き、必要なオプションを設定します。

環境構成の 「Dataplex データリネージ統合」 を編集します。

cloud-composer-dataplex_03

「Dataplex データリネージとの統合を有効にする」 を選択して保存します。

cloud-composer-dataplex_05

環境の更新が行われ、しばらくすると環境構成の 「Dataplex データリネージ統合」 が有効になりました。

DAG を作成する

Cloud Composer 環境でデータリネージ統合を有効にして、サポートされているオペレーター を利用する DAG を実行すると、Cloud Composer が Data Lineage API にリネージ情報を報告します。

サポートされているオペレーターについては、下記ドキュメントを参照してください。

サポートされている演算子

今回はGCSToBigQueryOperatorを使用してみました。
以下は、Google Cloud Storage (GCS) にあるファイルを BigQuery に取り込むための Airflow DAG の例です。

gcs_to_bigquery_lineage_dag.py

from airflow import DAG
from airflow.utils.dates import days_ago
from airflow.providers.google.cloud.transfers.gcs_to_bigquery import GCSToBigQueryOperator

PROJECT_ID = '{プロジェクトID}'
DATASET_ID = 'work'
TABLE_ID = 'sales_data'
BUCKET_NAME = 'cm_enokawa_work'

# DAGの基本設定
default_args = {
    'start_date': days_ago(1),
}

with DAG(
    dag_id='gcs_to_bigquery_lineage_dag',
    default_args=default_args,
    schedule_interval=None,  # 手動で実行する
    catchup=False
) as dag:

    # GCS から BigQuery へのデータ取り込み
    gcs_to_bq = GCSToBigQueryOperator(
        task_id='load_gcs_to_bq',
        bucket=BUCKET_NAME,  # GCS バケット名
        source_objects=['sales.csv'],  # GCS 内のファイルパス
        destination_project_dataset_table=f'{PROJECT_ID}.{DATASET_ID}.{TABLE_ID}',  # BigQuery のデータセットとテーブル
        source_format='CSV',  # ソースファイル形式
        write_disposition='WRITE_TRUNCATE',  # 既存データの上書き
        skip_leading_rows=1  # CSV のヘッダー行をスキップ
    )

    gcs_to_bq

この DAG では、Google Cloud Storage に保存された CSV ファイル (sales.csv) を BigQuery のテーブル（sales_data）にインポートしています。
GCSToBigQueryOperatorによるリネージの追跡は自動的に行われるため、特に追加の操作は必要ありません。